package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

object Demo20Partition {
  def main(args: Array[String]): Unit = {
    /**
     * shuffle类的算子 在使用的时候可以手动指定分区数
     */

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("Demo20Partition")

    // 通过conf设置Spark默认的并行度
    conf.set("spark.default.parallelism", "4")

    val sc: SparkContext = new SparkContext(conf)

    // 读取单词数据构建RDD
    // 如果在集群中运行的时候 默认最少是 2个分区
    val lineRDD: RDD[String] = sc.textFile("spark/data/words/", 2)
    println("lineRDD is " + lineRDD.getNumPartitions)

    // 如果没有产生shuffle 那么子RDD的分区数由父RDD的分区数决定
    val wordsRDD: RDD[String] = lineRDD.flatMap(_.split(","))
    println("wordsRDD is " + wordsRDD.getNumPartitions)

    val mapRDD: RDD[(String, Int)] = wordsRDD.map(word => (word, 1))
    println("mapRDD is " + mapRDD.getNumPartitions)

    // repartition 可以通过shuffle改变分区的数量
    // 相当于一个转换算子 但是不做任何逻辑上的处理
    val repRDD: RDD[(String, Int)] = mapRDD.repartition(4)
    println("repRDD is " + repRDD.getNumPartitions)

    // coalesce 也是一个转换算子，也可以调整分区数
    // 一般用于 减少分区数量 而且不会产生shuffle
    // 也可以接受一个参数 shuffle 默认是false
    // 如果shuffle = true 相当于 repartition
    // 分区规则默认是 hash分区
    val coaRDD: RDD[(String, Int)] = repRDD.coalesce(2)
    println("coaRDD is " + coaRDD.getNumPartitions)

    // shuffle类的算子可以手动调整分区数
    // 相当于手动设置reduce任务的个数
    val wordCntRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _, 3)
    println("wordCntRDD is " + wordCntRDD.getNumPartitions)

    /**
     * shuffle类算子产生的RDD的分区数决定因素：
     * 1、如果没有指定，默认等与父RDD的分区数
     * 2、可以手动指定修改分区数量
     * 3、通过默认参数设置 spark.default.parallelism
     *
     * 手动设置的 > 默认参数（spark.default.parallelism） > 父RDD的分区数
     */
    wordCntRDD.foreach(println)

    val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    linesRDD
      .groupBy((line: String) => line, new MyPartitioner)
      .map(kv=>{
        val line: String = kv._1
        line
      }).saveAsTextFile("spark/data/clazz_partition")

  }


}

/**
 * 将学生信息表的数据 分为两个分区
 * 文科一个、理科一个
 */
// 自定义分区器
class MyPartitioner extends Partitioner {
  // 分区数量
  override def numPartitions: Int = 2

  // 指定分区规则 默认：Hash分区
  override def getPartition(key: Any): Int = {
    val line: String = key.toString
    val clazz: String = line.split(",")(4)
    if (clazz.startsWith("文科"))
      0
    else
      1
  }
}
